package david.java.flink_sql.streamApiIntegration;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.DiscardingSink;
import org.apache.flink.table.api.DataTypes;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableDescriptor;
import org.apache.flink.table.api.bridge.java.StreamStatementSet;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @Description:
 * @See https://nightlies.apache.org/flink/flink-docs-release-1.14/docs/dev/table/data_stream_api/#adding-table-api-pipelines-to-datastream-api
 * @Author: ZhaoDawei
 * @Date: Create in 10:09 上午 2022/4/6
 */
public class S5_AddTableApi2DatastreamApi {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        StreamStatementSet statementSet = tableEnv.createStatementSet();

        TableDescriptor sourceDescriptor = TableDescriptor.forConnector("datagen")
                .option("number-of-rows", "3")
                .schema(
                        Schema.newBuilder()
                                .column("myCol", DataTypes.INT())
                                .column("myOtherCol", DataTypes.BOOLEAN()
                                ).build()
                )
                .build();

        TableDescriptor sinkDescriptor = TableDescriptor.forConnector("print").build();

        Table tableFromSource = tableEnv.from(sourceDescriptor);
        statementSet.addInsert(sinkDescriptor, tableFromSource);

        DataStreamSource<Integer> dataStream = env.fromElements(1, 2, 3);
        Table tableFromStream = tableEnv.fromDataStream(dataStream);
        statementSet.addInsert(sinkDescriptor, tableFromStream);



        // attach both pipelines to StreamExecutionEnvironment
        // (the statement set will be cleared after calling this method)
        statementSet.attachAsDataStream();


        // define other DataStream API parts
        env.fromElements(4, 5, 6).addSink(new DiscardingSink<>());

        // use DataStream API to submit the pipelines
        env.execute();

// prints similar to:
// +I[1618440447, false]
// +I[1259693645, true]
// +I[158588930, false]
// +I[1]
// +I[2]
// +I[3]


    }
}
